package wsz.order;

import com.alibaba.rocketmq.client.consumer.DefaultMQPullConsumer;
import com.alibaba.rocketmq.client.consumer.PullResult;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.alibaba.rocketmq.common.message.MessageQueue;

import java.util.List;
import java.util.Set;

/**
 * 部分有序消费
 * @author wsz
 * @date 2022/2/28 10:25
 **/
public class Consumer {

    public static void main(String[] args) throws Exception {
        DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("consumer_grp_wsz");
        consumer.setNamesrvAddr("192.168.174.138:9876");
        consumer.start();

        // 获取主题所有的message queue
        Set<MessageQueue> messageQueues = consumer.fetchSubscribeMessageQueues("tp_demo_08");

        for (MessageQueue messageQueue: messageQueues) {
            // 记录拉取下标
            long nextBeginOffset = 0;
            do {
                // 从指定message queue中拉取数据
                PullResult pullResult = consumer.pull(messageQueue, "*", nextBeginOffset, 1);
                if (pullResult == null || pullResult.getMsgFoundList() == null) break;;

                nextBeginOffset = pullResult.getNextBeginOffset();
                // 拉取到的数据
                List<MessageExt> msgFoundList = pullResult.getMsgFoundList();
                for (MessageExt messageExt : msgFoundList) {
                    System.out.println(messageExt);
                }
            } while (true);
        }
        consumer.shutdown();
    }
}
